
package hasthi.common;

import hasthi.container.p2p.XmlMesssage;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.xmlpull.v1.builder.XmlElement;

/**
 * Class handle the message interceptors and filters. Message interceptors stand before the service and process the requests
 * directed to service. Usually the pick the requests by Soap action. Filters allow messages to be intercepted and stored
 * in a internal map. Also This class allows users to wait for some Message to be arrived (waitFor(..)). On the wait, the thread goes to
 * sleep and it is not a busy wait. 
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class MessageRendezvous {
    private Map<String,XmlMesssage> messageMap = new HashMap<String, XmlMesssage>();
    private Map<String,Semaphore> semaphoreMap = new HashMap<String, Semaphore>();

    public void addMessage(XmlElement message){
        throw new UnsupportedOperationException();
    }
    
    /**
     * Add Message checks is there a semaphore register for this key, there is a semaphore
     * it will do a release for semaphore.
     * @param key
     * @param message
     */
    public final void addMessage(String key,XmlMesssage message){
        synchronized (messageMap) {
            Semaphore semaphore = semaphoreMap.get(key);
            messageMap.put(key, message);
            if(semaphore != null){
                semaphore.release();
            }
        }
    }
    
    
    /**
     * Method will hang till the message with given key is arrived. The code try to get the Message and if it is null, 
     * it will create a new semaphore, add it to the map and wait on it. When message is added, the thread will be released. 
     * @param id
     * @return
     * @throws GfacException
     */
    public final XmlMesssage waitFor(String id) throws InterruptedException{
       Semaphore semaphore = null;
       synchronized (messageMap) {
           XmlMesssage message = messageMap.get(id);
            if(message != null){
                return  message;
            }
            semaphore = new Semaphore(0,true);
            semaphoreMap.put(id, semaphore);
       }    
       semaphore.acquire();
       synchronized (messageMap) {
            semaphoreMap.remove(id);
            return messageMap.remove(id);
        }
    }
    
    
    /**
     * This is same as  waitFor(String id), but the wait time out after given time
     * @param id
     * @return
     * @throws GfacException
     */
    public final XmlMesssage waitFor(String id,int waittimeInSeconds) throws InterruptedException{
        try {
           Semaphore semaphore = null;
           synchronized (messageMap) {
               XmlMesssage message = messageMap.get(id);
                if(message != null){
                    return  message;
                }
                semaphore = new Semaphore(0,true);
                semaphoreMap.put(id, semaphore);
           }    
           semaphore.tryAcquire(waittimeInSeconds, TimeUnit.SECONDS);
           synchronized (messageMap) {
                semaphoreMap.remove(id);
                return messageMap.remove(id);
            }
        } catch (InterruptedException e) {
            //TODO clean up the semaphore
            throw e;
        }
    }
    
    
    public XmlMesssage getMessage(String id){
        synchronized (messageMap) {
            return messageMap.remove(id);
        }
    }
}

